package action

import (
	"strings"
	"time"

	"gitee.com/sansaniot/ssiot-core/facade/env"
	"gitee.com/sansaniot/ssiot-core/logger"
	"gitee.com/sansaniot/ssiot-core/ws"
	"gorm.io/gorm"

	"ssdevice/common/constant"
	"ssdevice/common/enums"
	"ssdevice/common/middleware/mqtt"
	"ssdevice/common/utils"
	"ssdevice/internal/device/models"
)

type Sensor struct {
	Orm *gorm.DB
}

func (action *Sensor) Execute(topic string, jsonData map[string]interface{}) {
	devSn := strings.Split(topic, "/")[3]
	_, to := AutoLogin(action.Orm, devSn, jsonData)
	if strings.HasSuffix(topic, "/sensor") {
		// 变量上报 - 主题去掉了post
		targetSn := utils.If(len(to) > 0 && to != devSn, to, devSn).(string)
		sensor := utils.GetMapValueFiled(jsonData, "data")
		now := utils.GetNowTimeStr()
		// 更新实时数据
		if sensor != nil {
			mqtt.DealData(targetSn, sensor, func(mData map[string]interface{}) {
				delete(mData, constant.CacheModel)
				action.SensorUpdate(SensorMsg{
					DevSn:  targetSn,
					Sensor: mData,
					Now:    now,
				})
			}, enums.SensorMsg)
		}
		// 更新dev状态
		UpdateLiveStatus(action.Orm, devSn, to, now)
	}
}

type SensorMsg struct {
	DevSn  string
	Sensor map[string]interface{}
	Now    string
}

var sensorQueue = constant.NewCQueue()

// 测量量sensor更新 (该方法在go func{}中调用)
func (action *Sensor) SensorUpdate(sensorMsg SensorMsg) {
	// 顺序消费
	sensorQueue.Enqueue(sensorMsg)
}

func init() {
	go func() {
		for {
			time.Sleep(100 * time.Millisecond)
			for {
				d := sensorQueue.Dequeue()
				if d == nil {
					break
				}
				data := d.(SensorMsg)
				addSensorQueue(data.DevSn, data.Sensor, data.Now)
				postProcess(data.DevSn, data.Sensor, enums.SensorMsg, data.Now)
			}
		}
	}()
}

func addSensorQueue(sn string, data map[string]interface{}, now string) {
	if len(data) == 0 {
		return
	}
	// 同时上报几条sensor时，第一个事务未完成设备初始，后面的sensor没进事务直接到此，导致SensorCache判断有问题
	for constant.InitializingDev.Has(sn) {
		time.Sleep(3 * time.Millisecond)
	}
	// 加入历史队列
	hisConsumer := constant.HisDataBucket[int(mqtt.N)%constant.DataRoutineChild]
	if hisConsumer != nil {
		hisData := models.SysHistorySensor{
			DevSn:            sn,
			Value:            utils.MapToBytes(data),
			DateTime:         now[0:10],
			UpdateTimeInsStr: now,
		}
		hisConsumer.Enqueue(hisData)
	}
	// 实时缓存处理
	var tmpMap map[string]interface{}
	if value, ok := constant.SensorCache.Get(sn); !ok {
		if err := utils.GobDeepCopy(data, &tmpMap); err != nil {
			logger.Error(err)
			return
		}
		// 插入队列
		constant.SensorInsCache.Set(sn, tmpMap)
	} else {
		if err := utils.GobDeepCopy(value, &tmpMap); err != nil {
			logger.Error(err)
			return
		}
		insMap := make(map[string]interface{})
		// 判断更新/插入
		for k, v := range data {
			if _, okk := tmpMap[k]; !okk {
				// 插入队列
				insMap[k] = v
			} else {
				// 哪些sn+key待更新(减少更新频次)
				constant.SensorSnKey.Set(sn+"#"+k, now)
			}
		}
		if len(insMap) > 0 {
			if toIns, ex := constant.SensorInsCache.Get(sn); ex && len(toIns) > 0 {
				// 同设备的待插入数据还未执行,则需要合并,否则被覆盖丢失
				utils.DeepMerge(toIns, insMap)
			}
			constant.SensorInsCache.Set(sn, insMap)
		}
		// 合并
		utils.DeepMerge(data, tmpMap)
	}
	// 再缓存
	constant.SensorCache.Set(sn, tmpMap)
}

// 后置处理
func postProcess(devSn string, data map[string]interface{}, msgType enums.MsgType, now string) {
	if len(data) == 0 {
		return
	}
	cacheData := map[string]interface{}{
		"sn":         devSn,
		"data":       data,
		"updateTime": now,
		"msgType":    msgType.Get().Name,
	}
	// 视频ai事件上报
	alarmEvent := utils.GetMapValueFiled(data, "alarm")
	if alarmEvent != nil && alarmEvent["event"] != nil {
		dptId, _ := env.Cache.HashGet(constant.DeviceDeptCache, devSn)
		cacheData["deptId"] = dptId
	}
	dataBytes := utils.MapToBytes(cacheData)
	// 广播已处理数据
	mqtt.MqttClient.Publish(constant.HandledMsgTopic, dataBytes)
	// 统计明细缓存
	addDataDetail(devSn, len(dataBytes))
	// websocket广播
	ws.BroadcastMsg(map[string][]byte{"device#" + devSn: dataBytes})
}

func addDataDetail(devSn string, dataSize int) {
	if tmpCount, ok := constant.DevDataCountCache.Get(devSn); !ok {
		constant.DevDataCountCache.Set(devSn, 1)
	} else {
		constant.DevDataCountCache.Set(devSn, tmpCount+1)
	}
	if tmpSize, ok := constant.DevDataSizeCache.Get(devSn); !ok {
		constant.DevDataSizeCache.Set(devSn, dataSize)
	} else {
		constant.DevDataSizeCache.Set(devSn, tmpSize+dataSize)
	}
}

// UpdateLiveStatus 更新活跃状态
func UpdateLiveStatus(db *gorm.DB, devSn, to, now string) {
	dev := &models.SysDev{}
	dev.Model.Orm = db
	if err := dev.UpdateOnlineStatus(devSn, to, now, enums.Online); err != nil {
		logger.Error(err)
	}
}
